package com.lagou.bak

import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext}
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object CheckpointDemo {

  def main(args: Array[String]): Unit = {
    //    1.获取env
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val source: DataStream[String] = env.socketTextStream("linux123", 7777)

    env.setParallelism(4)
    env.enableCheckpointing(5000)

    source.map(_.toInt)
      .addSink(new MySink)

    env.execute("test")

  }

  class MySink extends SinkFunction[Int] with CheckpointedFunction {

    override def snapshotState(functionSnapshotContext: FunctionSnapshotContext): Unit = {
      println("snapshotState...")
    }

    override def initializeState(functionInitializationContext: FunctionInitializationContext): Unit = {
      println("initializeState...")
      println(s"isRestored: ${functionInitializationContext.isRestored}")
      if (functionInitializationContext.isRestored) {
        println("===========程序出错，自动恢复执行！")
      }
    }

    override def invoke(value: Int, context: SinkFunction.Context[_]): Unit = {
      println(value)
    }
  }

}
